package com.atguigu.flink.streamapi.trans;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Created by Smexy on 2023/1/13
 *
 *  分流。
 *          生活中有主干道，还有辅路。
 *
 *          流有主流，还有侧流。
 *
 *     希望把s1传感器的数据放入主流，s2和s3都放入两个侧流。
 *
 *     -----------
 *     侧流使用场景，用于将区别于主流的数据，放入侧流。
 *      例如： 主流放的是非迟到的数据。
 *            迟到的数据，放入侧流处理
 */
public class Demo10_SideOutPut
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建OutputTag
        OutputTag<WaterSensor> s2OutputTag = new OutputTag<>("s2",TypeInformation.of(WaterSensor.class));
        OutputTag<WaterSensor> s3OutputTag = new OutputTag<>("s3",TypeInformation.of(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .process(new ProcessFunction<WaterSensor, WaterSensor>()
            {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                    if ("s1".equals(value.getId())) {
                        //主流输出
                        out.collect(value);
                    } else if ("s2".equals(value.getId())) {
                               /*
                                侧流输出
                                    OutputTag<X> outputTag: 输出标记。侧流中输出的是什么类型的数据，为侧流起一个名字
                                    X value: 输出的数据
                                */
                        ctx.output(s2OutputTag, value);
                    } else if ("s3".equals(value.getId())) {
                        ctx.output(s3OutputTag, value);
                    }

                }
            });

        //返回的流式主流  标准输出 黑色
        ds.print("主流");
        DataStream<WaterSensor> s2 = ds.getSideOutput(s2OutputTag);
        DataStream<WaterSensor> s3 = ds.getSideOutput(s3OutputTag);
        //向控制台的错误输出  红色
        s2.printToErr("s2");
        s3.printToErr("s3");


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
